﻿using System;
using System.Net.Sockets;
using System.Threading.Tasks;
using ePipe;
using eTask.Extensions;
using Letter.IO.Memorizer;
using Letter.Net.Sockets;

namespace Letter.Net.Session;

sealed partial class TcpSession<TSnders>
{
    sealed class TcpSessionRcver
    {
        private static readonly int MinAllocBufferSize = MemoryPoolProvider.GetBlockSize() / 2;
        public TcpSessionRcver(TcpSession<TSnders> channel)
        {
            this.channel = channel;
        }

        private TcpSession<TSnders> channel;
        private Task rcvTask;

        private IPipeWriter<byte> Input
        {
            get { return channel.Application.Writer; }
        }

        private IPipeReader<byte> Output
        {
            get { return channel.Application.Reader; }
        }

        private Socket Socket
        {
            get { return this.channel.socket; }
        }

        private TcpRcver Rcver
        {
            get { return this.channel.SocketRcver; }
        }

        public void Start()
        {
            this.rcvTask = this.StartRcvAsync();
        }

        private async Task StartRcvAsync()
        {
            var socketRcver = this.Rcver;
            var socket = this.Socket;
            var input = this.Input;

            Exception error = null;
            try
            {
                while (true)
                {
                    var waitResult = await socketRcver.WaitAsync(socket);
                    if (!SocketOperationUtil.TryNormalCompletion(waitResult, out error))
                    {
                        break;
                    }

                    var memory = input.GetMemory(MinAllocBufferSize);
                    var readResult = await socketRcver.ReceiveAsync(socket, memory);
                    if (!SocketOperationUtil.TryNormalCompletion(readResult, out error))
                    {
                        break;
                    }
                    
                    var bytesReceived = readResult.BytesTransferred;
                    if (bytesReceived == 0)
                    {
                        break;
                    }

                    input.Advance(bytesReceived);
                    
                    var flushTask = input.FlushAsync();
                    var flushResult = await flushTask;
                    if (flushResult.IsCanceled || flushResult.IsCompleted)
                    {
                        break;
                    }
                }
            }
            catch (ObjectDisposedException ex)
            {
                error = ex;
            }
            catch (Exception ex)
            {
                error = ex;
            }
            finally
            {
                if (!this.channel.closed)
                {
                    if (error != null)
                    {
                        this.channel.Connection.OnException(error);
                    }

                    this.channel.CloseAsync().Ignore();
                }
            }
        }

        public async ValueTask CloseAsync()
        {
            this.AbortSocketRcverIO();

            await rcvTask;
        }

        private void AbortSocketRcverIO(Exception error= null)
        {
            this.Input.Complete(error);
            this.Output.CancelPendingRead();
        }
    }
}